{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8ce67390",
   "metadata": {},
   "source": [
    "# MCP协议加持，千帆AppBuilder让语音导航更智能，开发更轻松\n",
    "\n",
    "MCP最近热度很高，作为国内首家支持MCP协议的大模型应用开发平台，千帆AppBuilder完成兼容后，用户可通过千帆AppBuilder轻松调用MCP Server生态中的海量工具，快速扩展端和云端的工具数量。我们的用户，也在利用这个强大的功能，打破不同工具间壁垒，实现更多想法。\n",
    "\n",
    "比如，爷爷奶奶都会用的百度地图，可以在千帆AppBuilder的连接下，成为你的开发利器：\n",
    "\n",
    "百度地图核心API已全面兼容MCP协议——你在千帆AppBuilder开发的应用，可以轻松调用百度地图的各项功能。对着手机说出“人话需求”，AI自动规划路线、检索地点、查天气、查路况。\n",
    "\n",
    "具体如何实现？本文将带你一步步了解如何通过千帆AppBuilder、百度地图和MCP协议的结合，通过SDK的方式进行开发，打造出更智能的应用。我们还会提供详细的安装和配置步骤，帮助你快速上手，轻松实现功能扩展。\n",
    "\n",
    "## 官方代码及文档\n",
    "\n",
    "- [百度地图 MCP Server](https://github.com/baidu-maps/mcp)\n",
    "- [Appbuilder 调用MCP组件服务](https://cloud.baidu.com/doc/AppBuilder/s/Um88v68dn)\n",
    "\n",
    "## 安装配置步骤\n",
    "\n",
    "### 配置基础环境\n",
    "\n",
    "#### 1、安装Python\n",
    "\n",
    "大部分的MCP Server，例如Unity和Blender，都是Python SDK\n",
    "安装Python最简单的方法是安装Miniconda\n",
    "**注意：请确保安装的python>=3.12**\n",
    "\n",
    "[Miniconda官网](https://www.anaconda.com/docs/getting-started/miniconda/install)\n",
    "\n",
    "#### 2、安装包管理器uv\n",
    "\n",
    "uv是pip的替代，能够快速启动和管理项目，后续每个MCP服务都需要用uvx命令启动\n",
    "\n",
    "[包管理器uv安装教程](https://docs.astral.sh/uv/getting-started/installation)\n",
    "\n",
    "#### 3、安装Node.js\n",
    "\n",
    "[Node.js安装教程](https://nodejs.org/en/download)\n",
    "\n",
    "#### 4、下载Appbuilder-SDK\n",
    "\n",
    "为了实现Appbuilder-SDK对百度地图 MCP Server的代码态调用，需要在conda环境下载appbuilder-sdk包\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "96a39a88",
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "!python3 -m pip install httpx appbuilder-sdk mcp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6e7d417d",
   "metadata": {},
   "source": [
    "#### 获取百度地图API Key\n",
    "\n",
    "在百度地图开放平台申请一个API Key，记下来\n",
    "[百度地图开放平台](https://lbsyun.baidu.com/apiconsole/key)\n",
    "\n",
    "## 百度智能云千帆AppBuilder代码态调用\n",
    "\n",
    "如何通过代码来调用百度地图的MCP服务，接入到我自己的应用中呢？\n",
    "\n",
    "可以用百度智能云企业级大模型应用开发平台-千帆AppBuidler的代码态，手搓一个MCP的客户端。\n",
    "\n",
    "### 1、进入百度智能云千帆AppBuilder主页\n",
    "\n",
    "进入百度智能云千帆AppBuilder主页，免费试用\n",
    "\n",
    "- [百度智能云千帆AppBuilder主页](https://qianfan.cloud.baidu.com/appbuilder)\n",
    "\n",
    "### 2、创建【自主规划Agent】\n",
    "\n",
    "- [Appbuilder 自主规划Agent概述](https://cloud.baidu.com/doc/AppBuilder/s/km4i50me8)\n",
    "\n",
    "- [快速创建应用](https://cloud.baidu.com/doc/AppBuilder/s/Om0ks82ow)\n",
    "\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E5%88%9B%E5%BB%BA%E8%87%AA%E4%B8%BB%E8%A7%84%E5%88%92Agent.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T15%3A00%3A03Z%2F-1%2Fhost%2F31df73d263dbc1dc7e28a6bc077b37c70ce27af91ac7888ff9151c7ef1d58b04)\n",
    "\n",
    "### 3、输入一句简单的提示词，让AI自动扩写\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E5%88%9B%E5%BB%BA%E8%87%AA%E4%B8%BB%E8%A7%84%E5%88%92agent.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T03%3A23%3A29Z%2F-1%2Fhost%2F67b7a5d0c38c12f79bc925e995c8f09775eb37bfe9025f61a593080c3c05c82c)\n",
    "\n",
    "### 4、删除自动配置的组件\n",
    "\n",
    "因为我们调用MCP工具，所以不需要使用AppBuilder自带组件\n",
    "\n",
    "### 5、配置模型\n",
    "\n",
    "将规划模型和问答模型的思考轮数拉满\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E9%85%8D%E7%BD%AE%E6%A8%A1%E5%9E%8B.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T03%3A27%3A44Z%2F-1%2Fhost%2Ffd031f86bf71a0ca7e32db70533a76db9c39757a139f94b34f487eae9f860534)\n",
    "\n",
    "### 6、发布应用\n",
    "\n",
    "发布应用，在【个人空间】中，查看应用ID，记下来\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E5%8F%91%E5%B8%83Agent%E5%BA%94%E7%94%A8.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T03%3A30%3A26Z%2F-1%2Fhost%2Ff66b1bef4bb432ddd59ca07189ed210fb475c6363559dfd0c7ab53ef47a6240f)\n",
    "\n",
    "### 7、创建Appbuilder API key\n",
    "\n",
    "- [Appbuilder API key授权](https://cloud.baidu.com/doc/AppBuilder/s/lm68r8e6i)\n",
    "\n",
    "- [Appbuilder API 控制台管理中心](https://console.bce.baidu.com/iam/#/iam/apikey/list)\n",
    "\n",
    "在【API Key】界面中，创建一个新的Key，并添加刚刚创建的应用，记下来。\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E5%88%9B%E5%BB%BAAPPBUILDER-TOKEN.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T14%3A49%3A27Z%2F-1%2Fhost%2F9266ac8c93a262d2905811877d680f53af26668a0ee09e1b3a61c0b721de253c)\n",
    "\n",
    "同时配置API key权限，将上文创建的应用添加到API key权限中。\n",
    "\n",
    "![](https://bj.bcebos.com/v1/appbuilder-sdk-components/%E6%B7%BB%E5%8A%A0Agent%E6%9D%83%E9%99%90.png?authorization=bce-auth-v1%2FALTAKGa8m4qCUasgoljdEDAzLm%2F2025-04-17T03%3A33%3A59Z%2F-1%2Fhost%2Feda8af9c241207cf07ff2edb85b2271ce48aade84775ca2273580ad13f03a5dc)\n",
    "\n",
    "### 8、配置代码\n",
    "\n",
    "将下述链接复制到浏览器，可下载代码。解压缩文件，并更新代码内容。\n",
    "\n",
    "https://qianfandeveloper.bj.bcebos.com/baidumap_mcp.zip\n",
    "\n",
    "* 在map.py中，将`BAIDU_MAPS_API_KEY`换成你的自己的百度地图API密钥。\n",
    "* 在test.py文件中，将`APPBUILDER_TOKEN`更换成自己的Apppbuilder API Key，并将`app_id`更换成自己的应用ID。\n",
    "\n",
    "接下来将提供一个完整的示例，演示如何使用百度智能云千帆AppBuilder的Agent调用百度地图MCP服务。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e05e6628",
   "metadata": {},
   "source": [
    "#### （1）创建MCP服务器实例\n",
    "\n",
    "* 将下述代码添加到本地`map.py`文件中，**注意，这里需要替换`BAIDU_MAPS_API_KEY`为你自己的百度地图API Key**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c32b3a7a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# map.py\n",
    "import os\n",
    "import copy\n",
    "import httpx\n",
    "from asyncio import sleep\n",
    "\n",
    "from mcp.server.fastmcp import FastMCP, Context\n",
    "\n",
    "# 创建MCP服务器实例\n",
    "mcp = FastMCP(\"mcp-server-baidu-maps\")\n",
    "# 设置API密钥，用于调用百度地图API，获取方式请参考：https://lbsyun.baidu.com/apiconsole/key\n",
    "os.environ[\"BAIDU_MAPS_API_KEY\"] = \"6IlHFTRGzNjPxxxxxxxxxxxxxxxxxx\"\n",
    "api_key = os.getenv(\"BAIDU_MAPS_API_KEY\")\n",
    "api_url = \"https://api.map.baidu.com\"\n",
    "\n",
    "\n",
    "def filter_result(data) -> dict:\n",
    "    \"\"\"\n",
    "    过滤路径规划结果，用于剔除冗余字段信息，保证输出给模型的数据更简洁，\n",
    "    避免长距离路径规划场景下chat中断\n",
    "    \"\"\"\n",
    "    # 创建输入数据的深拷贝以避免修改原始数据\n",
    "    processed_data = copy.deepcopy(data)\n",
    "\n",
    "    # 检查是否存在'result'键\n",
    "    if \"result\" in processed_data:\n",
    "        result = processed_data[\"result\"]\n",
    "\n",
    "        # 检查'result'中是否存在'routes'键\n",
    "        if \"routes\" in result:\n",
    "            for route in result[\"routes\"]:\n",
    "                # 检查每个'route'中是否存在'steps'键\n",
    "                if \"steps\" in route:\n",
    "                    new_steps = []\n",
    "                    for step in route[\"steps\"]:\n",
    "                        # 提取'instruction'字段，若不存在则设为空字符串\n",
    "                        new_step = {\n",
    "                            \"distance\": step.get(\"distance\", \"\"),\n",
    "                            \"duration\": step.get(\"duration\", \"\"),\n",
    "                            \"instruction\": step.get(\"instruction\", \"\"),\n",
    "                        }\n",
    "                        new_steps.append(new_step)\n",
    "                    # 替换原steps为仅含instruction的新列表\n",
    "                    route[\"steps\"] = new_steps\n",
    "\n",
    "    return processed_data\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_geocode(address: str, ctx: Context) -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        地理编码服务\n",
    "\n",
    "    Description:\n",
    "        将地址解析为对应的位置坐标。地址结构越完整，地址内容越准确，解析的坐标精度越高。\n",
    "\n",
    "    Args:\n",
    "        address: 待解析的地址。最多支持84个字节。可以输入两种样式的值，分别是：\n",
    "        1、标准的结构化地址信息，如北京市海淀区上地十街十号【推荐，地址结构越完整，解析精度越高】\n",
    "        2、支持\"*路与*路交叉口\"描述方式，如北一环路和阜阳路的交叉路口\n",
    "        第二种方式并不总是有返回结果，只有当地址库中存在该地址描述时才有返回。\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/geocoding/v3/\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbsyun.baidu.com/faq/api?title=webapi/guide/webservice-geocoding\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"address\": f\"{address}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_reverse_geocode(\n",
    "    latitude: float, longitude: float, ctx: Context\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        逆地理编码服务\n",
    "\n",
    "    Description:\n",
    "        根据经纬度坐标点获取对应位置的行政区划与POI信息\n",
    "\n",
    "    Args:\n",
    "        latitude: 纬度 (gcj02ll)\n",
    "        longitude: 经度 (gcj02ll)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"There\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/reverse_geocoding/v3/\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbsyun.baidu.com/faq/api?title=webapi/guide/webservice-geocoding-abroad\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"coordtype\": \"gcj02ll\",\n",
    "            \"location\": f\"{latitude},{longitude}\",\n",
    "            \"extensions_poi\": \"1\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_search_places(\n",
    "    query: str, region: str, location: str, radius: int, ctx: Context\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        地点检索服务\n",
    "\n",
    "    Description:\n",
    "        城市内检索: 检索某一城市内（目前最细到城市级别）的地点信息。\n",
    "        周边检索: 设置圆心和半径，检索圆形区域内的地点信息（常用于周边检索场景）。\n",
    "\n",
    "    Args:\n",
    "        query: 检索关键字\n",
    "        region: 检索的行政区划\n",
    "        location: 圆形区域检索中心点\n",
    "        radius: 圆形区域检索半径，单位：米\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/place/v2/search\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbsyun.baidu.com/faq/api?title=webapi/guide/webservice-placeapi\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"query\": f\"{query}\",\n",
    "            \"region\": f\"{region}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "        if location:\n",
    "            params[\"location\"] = f\"{location}\"\n",
    "            params[\"radius\"] = f\"{radius}\"\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_place_details(uid: str, ctx: Context) -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        地点详情检索服务\n",
    "\n",
    "    Description:\n",
    "        地点详情检索: 地点详情检索针对指定POI，检索其相关的详情信息。\n",
    "        开发者可以通地点检索服务获取POI uid。使用\"地点详情检索\"功能，传入uid，\n",
    "        即可检索POI详情信息，如评分、营业时间等（不同类型POI对应不同类别详情数据）。\n",
    "\n",
    "    Args:\n",
    "        uid: poi的唯一标识\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/place/v2/detail\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbsyun.baidu.com/faq/api?title=webapi/guide/webservice-placeapi/detail\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"uid\": f\"{uid}\",\n",
    "            # Agent入参不可控，这里给定scope为2\n",
    "            \"scope\": 2,\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_distance_matrix(\n",
    "    origins: str, destinations: str, mode: str, ctx: Context\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        批量算路服务\n",
    "\n",
    "    Description:\n",
    "        根据起点和终点坐标计算路线规划距离和行驶时间\n",
    "        批量算路目前支持驾车、骑行、步行\n",
    "        步行时任意起终点之间的距离不得超过200KM，超过此限制会返回参数错误\n",
    "        驾车批量算路一次最多计算100条路线，起终点个数之积不能超过100\n",
    "\n",
    "    Args:\n",
    "        origins: 多个起点坐标纬度,经度，按|分隔。示例：40.056878,116.30815|40.063597,116.364973【骑行】【步行】支持传入起点uid提升绑路准确性，格式为：纬度,经度;POI的uid|纬度,经度;POI的uid。示例：40.056878,116.30815;xxxxx|40.063597,116.364973;xxxxx\n",
    "        destinations: 多个终点坐标纬度,经度，按|分隔。示例：40.056878,116.30815|40.063597,116.364973【【骑行】【步行】支持传入终点uid提升绑路准确性，格式为：纬度,经度;POI的uid|纬度,经度;POI的uid。示例：40.056878,116.30815;xxxxx|40.063597,116.364973;xxxxx\n",
    "        mode: 批量算路类型(driving, riding, walking)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/routematrix/v2/{mode}\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbsyun.baidu.com/faq/api?title=webapi/routchtout\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"origins\": f\"{origins}\",\n",
    "            \"destinations\": f\"{destinations}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_directions(\n",
    "    model: str, origin: str, destination: str, ctx: Context\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        路线规划服务\n",
    "\n",
    "    Description:\n",
    "        驾车路线规划: 根据起终点坐标规划驾车出行路线\n",
    "        骑行路线规划: 根据起终点坐标规划骑行出行路线\n",
    "        步行路线规划: 根据起终点坐标规划步行出行路线\n",
    "        公交路线规划: 根据起终点坐标规划公共交通出行路线\n",
    "\n",
    "    Args:\n",
    "        model: 路线规划类型(driving, riding, walking, transit)\n",
    "        origin: 起点坐标，当用户只有起点名称时，需要先通过地理编码服务或地点地点检索服务确定起点的坐标\n",
    "        destination: 终点坐标，当用户只有起点名称时，需要先通过地理编码服务或地点检索服务确定起点的坐标\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/directionlite/v1/{model}\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbs.baidu.com/faq/api?title=webapi/direction-api-v2\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"origin\": f\"{origin}\",\n",
    "            \"destination\": f\"{destination}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        \"\"\"\n",
    "        过滤非公交的导航结果，防止返回的结果中包含大量冗余坐标信息，\n",
    "        影响大模型的响应速度，或是导致chat崩溃。\n",
    "        当前只保留导航结果每一步的距离、耗时和语义化信息。\n",
    "        公交路线规划情况比较多，尽量全部保留。\n",
    "        \"\"\"\n",
    "        if model == \"transit\":\n",
    "            return result\n",
    "        else:\n",
    "            return filter_result(result)\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_weather(location: str, district_id: int, ctx: Context) -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        天气查询服务\n",
    "\n",
    "    Description:\n",
    "        用户可通过行政区划或是经纬度坐标查询实时天气信息及未来5天天气预报\n",
    "        (注意: 使用经纬度坐标需要高级权限)。\n",
    "\n",
    "    Args:\n",
    "        location: 经纬度，经度在前纬度在后，逗号分隔 (需要高级权限, 例如: 116.30815,40.056878)\n",
    "        district_id: 行政区划 (例如: 1101010)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/weather/v1/?\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbs.baidu.com/faq/api?title=webapi/weather\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"data_type\": \"all\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        # 核心入参，二选一\n",
    "        if not location:\n",
    "            params[\"district_id\"] = f\"{district_id}\"\n",
    "        else:\n",
    "            params[\"location\"] = f\"{location}\"\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_ip_location(\n",
    "    # ip: str,\n",
    "    ctx: Context,\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        IP定位服务\n",
    "\n",
    "    Description:\n",
    "        根据用户请求的IP获取当前的位置，当需要知道用户当前位置、所在城市时可以调用该工具获取\n",
    "\n",
    "    Args:\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/location/ip\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbs.baidu.com/faq/api?title=webapi/ip-api\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_road_traffic(\n",
    "    model: str,\n",
    "    road_name: str,\n",
    "    city: str,\n",
    "    bounds: str,\n",
    "    vertexes: str,\n",
    "    center: str,\n",
    "    radius: int,\n",
    "    ctx: Context,\n",
    ") -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        实时路况查询服务\n",
    "\n",
    "    Description:\n",
    "        查询实时交通拥堵情况, 可通过指定道路名和区域形状(矩形, 多边形, 圆形)进行实时路况查询。\n",
    "\n",
    "        道路实时路况查询: 查询具体道路的实时拥堵评价和拥堵路段、拥堵距离、拥堵趋势等信息\n",
    "        矩形区域实时路况查询: 查询指定矩形地理范围的实时拥堵情况和各拥堵路段信息\n",
    "        多边形区域实时路况查询: 查询指定多边形地理范围的实时拥堵情况和各拥堵路段信息\n",
    "        圆形区域(周边)实时路况查询: 查询某中心点周边半径范围内的实时拥堵情况和各拥堵路段信息\n",
    "\n",
    "    Args:\n",
    "        model:      路况查询类型(road, bound, polygon, around)\n",
    "        road_name:  道路名称和道路方向, model=road时必传 (如:朝阳路南向北)\n",
    "        city:       城市名称或城市adcode, model=road时必传 (如:北京市)\n",
    "        bounds:     区域左下角和右上角的经纬度坐标, model=bound时必传 (如:39.912078,116.464303;39.918276,116.475442)\n",
    "        vertexes:   多边形区域的顶点经纬度, model=polygon时必传 (如:39.910528,116.472926;39.918276,116.475442;39.916671,116.459056;39.912078,116.464303)\n",
    "        center:     圆形区域的中心点经纬度坐标, model=around时必传 (如:39.912078,116.464303)\n",
    "        radius:     圆形区域的半径(米), 取值[1,1000], model=around时必传 (如:200)\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用百度API\n",
    "        url = f\"{api_url}/traffic/v1/{model}?\"\n",
    "\n",
    "        # 设置请求参数\n",
    "        # 更多参数信息请参考:https://lbs.baidu.com/faq/api?title=webapi/traffic\n",
    "        params = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"output\": \"json\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        # 核心入参，根据model选择\n",
    "        match model:\n",
    "            case \"bound\":\n",
    "                params[\"bounds\"] = f\"{bounds}\"\n",
    "            case \"polygon\":\n",
    "                params[\"vertexes\"] = f\"{vertexes}\"\n",
    "            case \"around\":\n",
    "                params[\"center\"] = f\"{center}\"\n",
    "                params[\"radius\"] = f\"{radius}\"\n",
    "            case \"road\":\n",
    "                params[\"road_name\"] = f\"{road_name}\"\n",
    "                params[\"city\"] = f\"{city}\"\n",
    "            case _:\n",
    "                pass\n",
    "\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            response = await client.get(url, params=params)\n",
    "            response.raise_for_status()\n",
    "            result = response.json()\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "        return result\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "@mcp.tool()\n",
    "async def map_poi_extract(text_content: str, ctx: Context) -> dict:\n",
    "    \"\"\"\n",
    "    Name:\n",
    "        POI智能提取\n",
    "\n",
    "    Description:\n",
    "        根据用户提供的文本描述信息, 智能提取出文本中所提及的POI相关信息.\n",
    "        (注意: 使用该服务, api_key需要拥有对应的高级权限, 否则会报错)\n",
    "\n",
    "    Args:\n",
    "        text_content: 用于提取POI的文本描述信息 (完整的旅游路线，行程规划，景点推荐描述等文本内容,\n",
    "                     例如: 新疆独库公路和塔里木湖太美了, 从独山子大峡谷到天山神秘大峡谷也是很不错的体验)\n",
    "    \"\"\"\n",
    "    # 关于高级权限使用的相关问题，请联系我们: https://lbsyun.baidu.com/apiconsole/fankui?typeOne=%E4%BA%A7%E5%93%81%E9%9C%80%E6%B1%82&typeTwo=%E9%AB%98%E7%BA%A7%E6%9C%8D%E5%8A%A1\n",
    "    try:\n",
    "        # 获取API密钥\n",
    "        if not api_key:\n",
    "            raise Exception(\"Can not found API key.\")\n",
    "\n",
    "        # 调用POI智能提取的提交接口\n",
    "        headers = {\"Content-Type\": \"application/x-www-form-urlencoded\"}\n",
    "        submit_url = f\"{api_url}/api_mark/v1/submit\"\n",
    "        result_url = f\"{api_url}/api_mark/v1/result\"\n",
    "\n",
    "        # 设置上传用户描述的请求体\n",
    "        submit_body = {\n",
    "            \"ak\": f\"{api_key}\",\n",
    "            \"id\": 0,\n",
    "            \"msg_type\": \"text\",\n",
    "            \"text_content\": f\"{text_content}\",\n",
    "            \"from\": \"py_mcp\",\n",
    "        }\n",
    "\n",
    "        # 异步请求\n",
    "        async with httpx.AsyncClient() as client:\n",
    "            # 提交任务\n",
    "            submit_resp = await client.post(\n",
    "                submit_url, data=submit_body, headers=headers, timeout=10.0\n",
    "            )\n",
    "            submit_resp.raise_for_status()\n",
    "            submit_result = submit_resp.json()\n",
    "\n",
    "            if submit_result.get(\"status\") != 0:\n",
    "                error_msg = submit_result.get(\"message\", \"unkown error\")\n",
    "                raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "            map_id = submit_result.get(\"result\", {}).get(\"map_id\")\n",
    "            if not map_id:\n",
    "                raise Exception(\"Can not found map_id\")\n",
    "\n",
    "            # 轮询获取结果（最多5次，间隔2秒）\n",
    "            result_body = {\n",
    "                \"ak\": api_key,\n",
    "                \"id\": 0,\n",
    "                \"map_id\": map_id,\n",
    "                \"from\": \"py_mcp\",\n",
    "            }\n",
    "            max_retries = 5\n",
    "            for attempt in range(max_retries):\n",
    "                result_resp = await client.post(\n",
    "                    result_url, data=result_body, headers=headers, timeout=10.0\n",
    "                )\n",
    "                result_resp.raise_for_status()\n",
    "                result = result_resp.json()\n",
    "\n",
    "                if result.get(\"status\") == 0 and result.get(\"result\"):\n",
    "                    return result\n",
    "                elif attempt < max_retries - 1:\n",
    "                    await sleep(2)\n",
    "\n",
    "            else:\n",
    "                raise Exception(\"Timeout to get the result\")\n",
    "\n",
    "        if result.get(\"status\") != 0:\n",
    "            error_msg = result.get(\"message\", \"unkown error\")\n",
    "            raise Exception(f\"API response error: {error_msg}\")\n",
    "\n",
    "    except httpx.HTTPError as e:\n",
    "        raise Exception(f\"HTTP request failed: {str(e)}\") from e\n",
    "    except KeyError as e:\n",
    "        raise Exception(f\"Failed to parse reponse: {str(e)}\") from e\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    mcp.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "65f88222",
   "metadata": {},
   "source": [
    "#### (2)配置Appbuilder自主规划Agent文件\n",
    "\n",
    "- 将下述代码添加到本地`test.py`文件中，**注意替换其中的`APPBUILDER_TOKEN`和`app_id`为你自己的AppBuilder的API key和应用ID**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52ec5f33",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import asyncio\n",
    "import appbuilder\n",
    "from appbuilder.core.console.appbuilder_client.async_event_handler import (\n",
    "    AsyncToolCallEventHandler,\n",
    ")\n",
    "from appbuilder.mcp_server.client import MCPClient\n",
    "\n",
    "# 用户Token\n",
    "os.environ[\"APPBUILDER_TOKEN\"] = \"bce-v3/ALTAK-mTvDIz1UXh1hXOXlPGhgZ/c248xxxxxxxxxxxxxxxxxxxxxxxx\"  # 将APPBUILDER_TOKEN更换成自己AppBuilder API key\n",
    "os.environ[\"APP_ID\"] = \"f1cd13f4-fec7xxxxxxxxxxxxxxxxxxxxxxxx\"  # 将app_id更换成自己的应用ID\n",
    "\n",
    "async def main():\n",
    "    # 发布的应用ID\n",
    "    app_id = os.environ[\"APP_ID\"]\n",
    "    appbuilder_client = appbuilder.AsyncAppBuilderClient(app_id)\n",
    "    mcp_client = MCPClient()\n",
    "\n",
    "    # 注意这里的路径为MCP Server文件在本地的相对路径\n",
    "    await mcp_client.connect_to_server(\"./map.py\")\n",
    "    \n",
    "    tools = mcp_client.tools\n",
    "    event_handler = AsyncToolCallEventHandler(mcp_client, functions=[])\n",
    "    conversation_id = await appbuilder_client.create_conversation()\n",
    "    \n",
    "    async with appbuilder_client.run_with_handler(\n",
    "        conversation_id=conversation_id,\n",
    "        query=\"我住在北京亮马河大厦，请帮我查询一下我当前的经纬度信息\",\n",
    "        tools=tools,\n",
    "        event_handler=event_handler,\n",
    "    ) as run:\n",
    "        await run.until_done()\n",
    "    \n",
    "    await appbuilder_client.http_client.session.close()\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    appbuilder.logger.setLoglevel(\"DEBUG\")\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(main())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5e6c0e92",
   "metadata": {},
   "source": [
    "### 9、运行代码文件\n",
    "\n",
    "在完成本地map.py文件和test.py文件内容更新后，在命令行中运行test.py，即可完成Agent的本地运行，控制台将呈现和上述结果一致的Agent运行结果"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f0b632dc",
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "!python3 test.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f02c8573",
   "metadata": {},
   "source": [
    "运行后，打印日志参考：\n",
    "```bash\n",
    "[2025-04-17 21:50:38,955.955] async_event_handler.py [line:563] DEBUG [main-11524059179509233548] Agent 非流式回答: 根据您的位置信息，北京亮马河大厦的经纬度为：**北纬39.95139408407192，东经116.47114501090634**。如果您有其他出行需求或问题，请随时告诉我，我会尽力为您提供帮助。\n",
    "DEBUG:appbuilder:Agent 非流式回答: 根据您的位置信息，北京亮马河大厦的经纬度为：**北纬39.95139408407192，东经116.47114501090634**。如果您有其他出行需求或问题，请随时告诉我，我会尽力为您提供帮助。\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f694b20",
   "metadata": {},
   "source": [
    "## 常用指令\n",
    "\n",
    "* 现在上海中山公园附近堵车吗\n",
    "* 今天上海辰山植物园天气怎么样\n",
    "* 我住在北京亮马河大厦，请帮我规划北京赏花一日游\n",
    "* 从上海环球港，开车导航到同济大学四平路校区"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
